package com.yy.smartBI.mqtest;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;

public class FanOutConsumer1 {
    private final static String QUEUE_NAME1 = "hello1";
    private final static String QUEUE_NAME2 = "hello2";


    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel1 = connection.createChannel();
        Channel channel2 = connection.createChannel();

        //声明交换机
        channel1.exchangeDeclare("logs", "topic");
        //声明队列
        channel1.queueDeclare(QUEUE_NAME1, true, false, false, null);
        //绑定队列到交换机
        channel1.queueBind(QUEUE_NAME1, "logs", "center.#");

        //声明交换机 同上
        channel2.exchangeDeclare("logs", "topic");
        channel2.queueDeclare(QUEUE_NAME2, true, false, false, null);
        channel2.queueBind(QUEUE_NAME2, "logs", "*.man");


        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x]heelo1 Received1 '" + message + "'");
            try {
                //模拟消息处理
                Thread.sleep(15000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        };

        DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x]heelo2 Received1 '" + message + "'");
            try {
                //模拟消息处理
                Thread.sleep(15000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        };

        channel1.basicConsume(QUEUE_NAME1, true, deliverCallback1, consumerTag -> { });
        channel2.basicConsume(QUEUE_NAME2, true, deliverCallback2, consumerTag -> { });
    }
}
